package com.woniu.servicesse.service.impl;

import com.woniu.internalcommon.dto.ResponseResult;
import com.woniu.servicesse.entity.Passenger;
import com.woniu.servicesse.feignClient.PassengerFeignClient;
import com.woniu.servicesse.request.PushRequest;
import com.woniu.servicesse.service.SseService;
import com.woniu.servicesse.util.SseConnection;
import com.woniu.servicesse.util.SsePrefixUtils;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.Synchronized;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

@Service
@Slf4j
public class SseServiceImpl implements SseService {
    @Autowired
    private PassengerFeignClient passengerFeignClient;
    //使用线程安全的map
    //key：用户ID+身份类型
    public static Map<String, SseConnection> sseEmitterMap = new ConcurrentHashMap<>();
    //创建一个map，记录最后一次活跃时间
    public static Map<String, Long> lastActiveTimeMap = new ConcurrentHashMap<>();
    private static final long HEARTBEAT_INTERVAL = 25_000; // 心跳间隔25秒
    private static final long CONNECTION_TIMEOUT = 90_000; // 连接超时时间15分钟
    private static final int HEARTBEAT_BATCH_SIZE = 100; // 每批处理心跳的连接数
    //创建线程池ThreadPoolExecutor
    // 线程池配置
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    private final ExecutorService taskExecutor = new ThreadPoolExecutor(
            4, // 核心线程数
            8, // 最大线程数
            60, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1000),
            new ThreadPoolExecutor.CallerRunsPolicy() // 由调用线程处理任务当队列满
    );
//    @Autowired
//    private RedissonClient redissonClient;
@PostConstruct  //bean初始化后执行
public void init() {
    // 启动心跳任务
    scheduler.scheduleAtFixedRate(
            new Runnable() {
                @Override
                public void run() {
                    scheduleHeartbeatTasks();
                }
            },
            HEARTBEAT_INTERVAL,
            HEARTBEAT_INTERVAL,
            TimeUnit.MILLISECONDS
    );

    // 启动连接超时检查任务
    scheduler.scheduleAtFixedRate(
            new Runnable() {
                @Override
                public void run() {
                    checkInactiveConnections();
                }
            },
            CONNECTION_TIMEOUT / 4, // 检查频率为超时时间的1/4
            CONNECTION_TIMEOUT / 4,
            TimeUnit.MILLISECONDS
    );
}
    //单线程
//    public void init() {
//        new Thread(() -> {
//            try {
//                Thread.sleep(CONNECTION_TIMEOUT); // 首次延迟
//                while (true) {
//                    checkInactiveConnections();
//                    Thread.sleep(CONNECTION_TIMEOUT); // 后续间隔
//                }
//            } catch (InterruptedException e) {
//                log.error("监控线程被中断", e);
//            }
//        }).start();
//    }
    @PreDestroy  // bean销毁时执行
    public void destroy() {
        scheduler.shutdownNow();
        taskExecutor.shutdown();
    }
    // 检测并关闭不活跃连接
    private void checkInactiveConnections() {
        long currentTime = System.currentTimeMillis();
        Set<String> keys = new HashSet<>(sseEmitterMap.keySet());

        for (String key : keys) {
            Long lastActiveTime = lastActiveTimeMap.get(key);
            SseConnection connection = sseEmitterMap.get(key);

            // 检查连接是否应该被关闭
            if (connection == null || lastActiveTime == null ||
                    currentTime - lastActiveTime > CONNECTION_TIMEOUT) {

                log.warn("关闭不活跃链接: {}", key);
                cleanupConnection(key);
            }
        }
    }

    //心跳机制，防止断连
    // 分派心跳任务到线程池
    private void scheduleHeartbeatTasks() {
        List<String> keys = new ArrayList<>(sseEmitterMap.keySet());

        // 分批发送心跳
        for (int i = 0; i < keys.size(); i += HEARTBEAT_BATCH_SIZE) {
            int end = Math.min(i + HEARTBEAT_BATCH_SIZE, keys.size());
            List<String> batchKeys = keys.subList(i, end);

            taskExecutor.execute(() -> {
                batchKeys.forEach(key -> {
                    // 再次检查连接是否存在，因为可能已被清理
                    if (sseEmitterMap.containsKey(key)) {
                        sendHeartbeatToConnection(key);
                    }
                });
            });
        }
    }

    // 给单个连接发送心跳
    private void sendHeartbeatToConnection(String key) {
        SseConnection connection = sseEmitterMap.get(key);
        if (connection == null || connection.isCompleted()) {
            return;
        }
        try {
            connection.getEmitter().send(SseEmitter.event()
                    .data("HEARTBEAT")
                    .comment("Keep-alive"));
            log.trace("心跳发送成功: {}", key);
        } catch (IllegalStateException e) {
            // 连接已经关闭
            log.debug("连接已关闭，跳过心跳: {}", key);
            cleanupConnection(key);
        } catch (Exception e) {
            log.warn("心跳发送失败: {}", key, e);
            cleanupConnection(key);
        }
    }

    // 清理连接资源
    private void cleanupConnection(String key) {
        SseConnection connection = sseEmitterMap.get(key);
        if (connection == null) {
            return;
        }
        try {
            if (!connection.isCompleted()) {
                try {
                    connection.getEmitter().send(SseEmitter.event().data("SSE_CLOSED"));
                } catch (Exception e) {
                    log.debug("发送关闭通知失败，连接可能已终止: {}", key);
                }
                connection.complete();
            }
        } finally {
            sseEmitterMap.remove(key);
            lastActiveTimeMap.remove(key);
        }
    }


    @Override
    @Synchronized
    public SseEmitter connect(String phone, Integer identity) {
        //获取的用户信息
        log.info("用户电话号码：{}；用户身份：{}", phone, identity);
        //根据用户电话号码查看用户id
        Long userId = null;
        String identityStr = null;
        if (identity == 1) {
            log.info("用户身份为：乘客");
            Map<String, Object> passengerInfo = (Map<String, Object>) passengerFeignClient.getUser(phone).getData();
            userId = (Long) passengerInfo.get("id");
            log.info("用户id：{}", userId);
            identityStr = "passenger";
        }
        log.info("进入sse连接方法");
        String sseMapKey = SsePrefixUtils.generatorSseKey(userId, identityStr);

        // 如果已有连接，先关闭旧的
        cleanupConnection(sseMapKey);

        SseEmitter emitter = new SseEmitter(0L);
        SseConnection connection = new SseConnection(emitter);

        // 初始化最后活跃时间
        lastActiveTimeMap.put(sseMapKey, System.currentTimeMillis());
        sseEmitterMap.put(sseMapKey, connection);

        try {
            // 发送初始数据
            emitter.send(SseEmitter.event().data("SSE_CONNECTED"));
            // 设置回调
            emitter.onCompletion(() -> cleanupConnection(sseMapKey));
            emitter.onTimeout(() -> cleanupConnection(sseMapKey));
            emitter.onError(e -> cleanupConnection(sseMapKey));
        } catch (IOException e) {
            cleanupConnection(sseMapKey);
            throw new RuntimeException("SSE连接建立失败", e);
        }
        return emitter;
    }

    @Override
    @Synchronized
    public String push(PushRequest pushRequest) {
        Long userId = pushRequest.getUserId();
        String identity = pushRequest.getIdentity();
        String content = pushRequest.getContent();
        log.info("用户ID：{}, 身份：{}", userId, identity);

        String sseMapKey = SsePrefixUtils.generatorSseKey(userId, identity);
        SseConnection connection = sseEmitterMap.get(sseMapKey);

        if (connection == null || connection.isCompleted()) {
            return "推送失败，连接不存在或已关闭";
        }

        // 更新最后活跃时间
        lastActiveTimeMap.put(sseMapKey, System.currentTimeMillis());

        // 使用线程池异步发送
        taskExecutor.execute(() -> {
            try {
                synchronized (connection) {
                    if (!connection.isCompleted()) {
                        connection.getEmitter().send(SseEmitter.event().data(content));
                        log.debug("推送成功: {}", sseMapKey);
                    } else {
                        log.debug("连接已关闭，跳过推送: {}", sseMapKey);
                        cleanupConnection(sseMapKey);
                    }
                }
            } catch (IllegalStateException e) {
                log.warn("连接已关闭，无法推送: {}", sseMapKey);
                cleanupConnection(sseMapKey);
            } catch (IOException e) {
                log.warn("推送失败: {}", sseMapKey, e);
                cleanupConnection(sseMapKey);
            }
        });

        return "给用户：" + sseMapKey + "，发送了消息：" + content;
    }

    @Override
    @Synchronized
    public String close(Long userId, String identity) {
        String sseMapKey = SsePrefixUtils.generatorSseKey(userId,identity);
//        RLock closeLock = redissonClient.getLock("sse:lock:" + sseMapKey);
//        System.out.println("关闭连接："+sseMapKey);
        cleanupConnection(sseMapKey);
        return "关闭连接成功";
    }
}
